package cn.com.daimajiangxin.flink;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * 简单的反序列化器：将 Kafka 消息（UTF-8 字符串）解析为 TaxiFare。
 * 支持 CSV 格式：driverId,eventTimeMillis,tip
 */
public class TaxiFareDeserializer implements DeserializationSchema<TaxiFare> {
    private static final long serialVersionUID = 1L;

    @Override
    public TaxiFare deserialize(byte[] message) throws IOException {
        if (message == null)
            return null;
        String s = new String(message, StandardCharsets.UTF_8).trim();
        if (s.isEmpty())
            return null;
        // CSV: driverId,eventTimeMillis,tip
        String[] parts = s.split(",");
        if (parts.length < 3) {
            // 简单容错：返回 null，记录日志可在生产环境中实现
            return null;
        }
        try {
            long driverId = Long.parseLong(parts[0].trim());
            long eventTime = Long.parseLong(parts[1].trim());
            float tip = Float.parseFloat(parts[2].trim());
            return new TaxiFare(driverId, eventTime, tip);
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public boolean isEndOfStream(TaxiFare nextElement) {
        return false;
    }

    @Override
    public TypeInformation<TaxiFare> getProducedType() {
        return TypeInformation.of(new TypeHint<TaxiFare>() {
        });
    }
}
